package com.it.stream;

import com.it.operator.utils.SourceUtils;
import com.it.pojo.Event;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.time.Duration;

/**
 * 一般情况下我们使用side output的方式进行分流操作，而且这些流的数据类型是可以不同的，每一个side output都
 * 可以定义为OutputTag[X]对象。processFunction可以通过context对象发射一个事件或者多个side outputs.
 *
 * @author code1997
 */
public class SplitStreamTest {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment executionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment();
        SingleOutputStreamOperator<Event> eventSource = SourceUtils.getEventSource(executionEnvironment).assignTimestampsAndWatermarks(WatermarkStrategy.<Event>forBoundedOutOfOrderness(Duration.ZERO).withTimestampAssigner(new SerializableTimestampAssigner<Event>() {
            @Override
            public long extractTimestamp(Event element, long recordTimestamp) {
                return element.timestamp;
            }
        }));
        //如果用户是tom，那么将数据输出到
        OutputTag<Tuple3<String, String, Long>> tomTag = new OutputTag<>("tom", Types.TUPLE(Types.STRING, Types.STRING, Types.LONG));
        SingleOutputStreamOperator<Event> processedStream = eventSource.process(new ProcessFunction<Event, Event>() {
            @Override
            public void processElement(Event value, ProcessFunction<Event, Event>.Context ctx, Collector<Event> out) throws Exception {
                if ("tom".equals(value.user)) {
                    ctx.output(tomTag, Tuple3.of(value.user, value.url, value.timestamp));
                } else {
                    out.collect(value);
                }
            }
        });
        processedStream.getSideOutput(tomTag).print("tom stream");
        processedStream.print("other user stream");
        executionEnvironment.execute();
    }

}
